package com.atguigu.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.ExamQuesqionBean;
import com.atguigu.util.ClickHouseUtil;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTestExamQuestionCount {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */

        //TODO 3 从topic_db中获取数据
        String topicDb = "topic_db";
        String groupId = "dws_test_exam_question_count";
        DataStreamSource<String> dataStreamSource = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicDb, groupId));
        //TODO 4 过滤脏数据
        SingleOutputStreamOperator<JSONObject> jsonObjStream = dataStreamSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    String tableName = jsonObject.getString("table");
                    String type = jsonObject.getString("type");
                    if ("test_exam_question".equals(tableName) && "insert".equals(type)) {
                        JSONObject data = jsonObject.getJSONObject("data");
                        data.put("ts",jsonObject.getLong("ts"));
                        out.collect(data);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    System.out.println("json格式转换错误");
                }
            }
        });

        //TODO 5 转换javabean
        SingleOutputStreamOperator<ExamQuesqionBean> beanStream = jsonObjStream
                .keyBy(value -> value.getString("id"))
                .flatMap(new RichFlatMapFunction<JSONObject, ExamQuesqionBean>() {
                    //记录独立用户
                    ValueState<String> lastExamState = null;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> lastExamDescriptor = new ValueStateDescriptor<>("lastExamDt", String.class);
                        lastExamDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).build());
                        lastExamState = getRuntimeContext().getState(lastExamDescriptor);
                    }

                    @Override
                    public void flatMap(JSONObject value, Collector<ExamQuesqionBean> out) throws Exception {
                        String lastExamDt = lastExamState.value();
                        long ts = value.getLong("ts") * 1000L;
                        String examDt = DateFormatUtil.toDate(ts);
                        //每条数据代表一个用户作答
                        Long answerTimeCt = 1L;
                        Long answerRightTimeCt = 0L;
                        Long answerRightUvCt = 0L;
                        Long answerUvCt = 0L;
                        if ("1".equals(value.getString("is_correct"))){
                            //如果该用户回答正确,正确数+1
                            answerRightTimeCt=1L;
                        }
                        if (lastExamDt==null || !lastExamDt.equals(examDt)){
                            //新数据是当日首次访问(独立作答用户+1)
                            answerUvCt=1L;
                            if ("1".equals(value.getString("is_correct"))){
                                //独立用户回答正确+1
                                answerRightUvCt=1L;
                            }
                            lastExamState.update(examDt);
                        }
                        out.collect(ExamQuesqionBean.builder()
                                .id(value.getString("id"))
                                .answerTimeCt(answerTimeCt)
                                .answerRightTimeCt(answerRightTimeCt)
                                .answerUvCt(answerUvCt)
                                .answerRightUvCt(answerRightUvCt)
                                .ts(value.getLong("ts")*1000L)
                                .build());

                    }
                });
        //TODO 6 设置水印开窗聚合
        SingleOutputStreamOperator<ExamQuesqionBean> reduceStream = beanStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<ExamQuesqionBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                .withTimestampAssigner(new SerializableTimestampAssigner<ExamQuesqionBean>() {
                    @Override
                    public long extractTimestamp(ExamQuesqionBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                })
        ).keyBy(value -> value.getId())
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)))
                .reduce(new ReduceFunction<ExamQuesqionBean>() {
                            @Override
                            public ExamQuesqionBean reduce(ExamQuesqionBean value1, ExamQuesqionBean value2) throws Exception {
                                value1.setAnswerRightUvCt(value1.getAnswerRightUvCt() + value2.getAnswerRightTimeCt());
                                value1.setAnswerUvCt(value1.getAnswerUvCt() + value2.getAnswerUvCt());
                                value1.setAnswerRightTimeCt(value1.getAnswerRightTimeCt() + value2.getAnswerRightTimeCt());
                                value1.setAnswerTimeCt(value1.getAnswerTimeCt() + value2.getAnswerTimeCt());
                                return value1;
                            }
                        },
                        new WindowFunction<ExamQuesqionBean, ExamQuesqionBean, String, TimeWindow>() {
                            @Override
                            public void apply(String s, TimeWindow window, Iterable<ExamQuesqionBean> input, Collector<ExamQuesqionBean> out) throws Exception {
                                ExamQuesqionBean next = input.iterator().next();
                                next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                                next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                                next.setTs(System.currentTimeMillis());
                                out.collect(next);
                            }
                        });
        //TODO 7 写入clickhouse
        reduceStream.addSink(ClickHouseUtil.getClickHouseSink("insert into dws_test_exam_question_count_window values(?,?,?,?,?,?,?,?)"));
        //TODO 8 启动
        env.execute(groupId);

    }
}
